package com.atguigu.chapter11;

import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.table.api.bridge.java.StreamTableEnvironment;
import org.apache.flink.table.catalog.hive.HiveCatalog;

/**
 * @Author lizhenchao@atguigu.cn
 * @Date 2021/6/19 9:26
 */
public class Flink10_Hive {
    public static void main(String[] args) throws Exception {
        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
        env.setParallelism(1);
        
        StreamTableEnvironment tEnv = StreamTableEnvironment.create(env);
        
        // 1. 创建hivecatalog
        HiveCatalog hive = new HiveCatalog("my_hive", "flink0107", "input");
        // 2. 注册catalog
        tEnv.registerCatalog("my_hive", hive);
        
        // 3. 设置要使用的catalog
        tEnv.useCatalog("my_hive");
        tEnv.useDatabase("flink0107");
        
        // 3. 直接从hive中查询数据
        //        tEnv.sqlQuery("select * from my_hive.flink0107.person").execute().print();
        tEnv.sqlQuery("select * from person").execute().print();
        
    }
}
